package com.ebs.md.fxmonitor.marketmonitor;

/**
 * Revision Author:
 *  $File: //depot/MarketData/fxmonitor/main/src/main/java/com/ebs/md/fxmonitor/marketmonitor/AbstractMonitorListenerAndPublisher.java $
 *  $Revision: #7 $
 *  $DateTime: 2012/05/24 08:52:58 $
 *  $Change: 111262 $
 *  $Author: Arik.Sher $
 */

import java.util.concurrent.atomic.AtomicLong;

import org.springframework.jms.core.JmsOperations;
import org.springframework.jms.core.MessageCreator;
import org.springframework.util.Assert;

import com.ebs.commons.comp.IParsedPayload;
import com.ebs.market.monitor.MarketMonitor.EventType;
import com.ebs.md.fxmonitor.marketmonitor.config.IEventConfigProvider;


public abstract class AbstractMonitorListenerAndPublisher
extends AbstractMonitorListener
{
	protected IEventConfigProvider eventConfigProvider;
	protected JmsOperations        jmsOperations; // used for publishing the Events
	protected String               jmsEventPublishDestinationName; // publicationDetination
	
	protected AtomicLong           eventSequence4log = new AtomicLong(1000000);
	
	public void init() {
		Assert.notNull(eventConfigProvider,                        "'eventConfigProvider' should not be null");
		Assert.notNull(jmsOperations,                                    "'jmsOperations' should not be null");
		Assert.notNull(jmsEventPublishDestinationName,  "'jmsEventPublishDestinationName' should not be null");
	}
	
	public void setEventConfigProvider(IEventConfigProvider eventConfigProvider) {
		this.eventConfigProvider = eventConfigProvider;
	}

	public void setJmsOperations(JmsOperations jmsOperations) {
		this.jmsOperations = jmsOperations;
	}

	public void setJmsEventPublishDestinationName(
			String jmsEventPublishDestinationName) {
		this.jmsEventPublishDestinationName = jmsEventPublishDestinationName;
	}

	protected void handleParsedPayload(IParsedPayload payload) {
		boolean toPublish = thresholdExceeds(payload);
		
		if ( toPublish ) {
			doPublish(payload);
		}
	}
    
	/**
	 * publishing 
	 * 
	 * @param payload input for creator
	 * @param creator specific to be used before sending
	 */
	protected void doPublish(IParsedPayload payload, EventMessageCreator creator) {
		long eventSeq4log = eventSequence4log.addAndGet(1);
		String payloadStr4log = payload.toString();
		logger.info(String.format("doPublish:: publishing %s event(seq=%d). payload: %s", creator.getEventType(), eventSeq4log, payloadStr4log));
		
		// publishing the JMS message
		jmsOperations.send(jmsEventPublishDestinationName, creator);
		
		if ( logger.isDebugEnabled() ) {
			logger.debug(String.format( "doPublish:: published %s event(seq=%d).", creator.getEventType(), eventSeq4log));
		} 
//		else {
//			if ( logger.isInfoEnabled() ) {
//				logger.info (String.format("doPublish:: published  %s event(seq=%d).", creator.getEventType(), eventSeq4log));
//			}
//		}
	}
	
	/**
	 * using default provided messageCreator on the input payload
	 * 
	 * @param payload input for creator
	 */
	protected void doPublish(IParsedPayload payload) {
		doPublish(payload, getCreator(payload));
	}
	
	/**
	 * tests whether the input exceeds it's threshold. 
	 * and should be published.
	 * 
	 * @param payload
	 * @return whether threshold exceeds 
	 */
	protected abstract boolean thresholdExceeds (IParsedPayload payload);

	protected abstract EventMessageCreator getCreator(IParsedPayload payload);
	
	
	static interface EventMessageCreator
	extends               MessageCreator
	{
		public EventType getEventType();		
	}
	
}
